package com.kenliang.netty.rpc.netty;

import com.kenliang.netty.rpc.core.MessageCallBack;
import com.kenliang.netty.rpc.model.MessageRequest;
import com.kenliang.netty.rpc.model.MessageResponse;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;

import java.util.concurrent.ConcurrentHashMap;

/**
 * @author KenLiang
 * @description:
 * @date Created in 5:46 PM 2019/3/21.
 */
@Slf4j
public class MessageSendHandler extends ChannelInboundHandlerAdapter {



    // todo 一个channel只会存在于一个线程中，感觉没有必要加这个volatile
//    private volatile Channel  channel;
    private  Channel  channel;
    private ConcurrentHashMap<String,MessageCallBack> callBackConcurrentHashMap = new ConcurrentHashMap<>();




    public void sendRequest(MessageRequest messageRequest){
        MessageCallBack messageCallBack = new MessageCallBack(messageRequest);
        callBackConcurrentHashMap.put(messageRequest.getMessageId(),messageCallBack);
        channel.writeAndFlush(messageRequest);
        messageCallBack.start();

    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        this.channel=ctx.channel();
        ctx.fireChannelRegistered();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // todo 接收response处理
        MessageResponse messageResponse = (MessageResponse)msg;
        MessageCallBack messageCallBack = callBackConcurrentHashMap.get(messageResponse.getMessageId());
        if(messageCallBack!=null){
            messageCallBack.over(messageResponse);
        }

        ctx.fireChannelRead(msg);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("MessageSendHandler exception:", ExceptionUtils.getStackTrace(cause));
        ctx.close();
    }
}
